package com.azarias.dfs.datanode.server;

import com.azarias.dfs.namenode.rpc.model.HeartbeatRequest;
import com.azarias.dfs.namenode.rpc.model.HeartbeatResponse;
import com.azarias.dfs.namenode.rpc.model.RegisterRequest;
import com.azarias.dfs.namenode.rpc.model.RegisterResponse;
import com.azarias.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyChannelBuilder;

import java.util.concurrent.CountDownLatch;

/**
 * 负责跟一组NameNode中的某一个进行通信的线程组件
 * @author Azarias_Jx
 *
 */
public class NameNodeServiceActor {
	private NameNodeServiceGrpc.NameNodeServiceBlockingStub namenode_rpc;

	//注册服务器的域名和端口用以rpc调用
	private final static String NAMENODE_HOSTNAME="localhost";
	private final static Integer NAMENODE_PORT=50070;
	/**
	 * 向自己负责通信的那个NameNode进行注册
	 */
	public void register( ) throws InterruptedException {
		Thread registerThread = new RegisterThread();
		registerThread.start();
		registerThread.join();
	}

	public NameNodeServiceActor() {
		ManagedChannel channel = NettyChannelBuilder
				.forAddress(NAMENODE_HOSTNAME, NAMENODE_PORT)
				.negotiationType(NegotiationType.PLAINTEXT)
				.build();
		this.namenode_rpc = NameNodeServiceGrpc.newBlockingStub(channel);
	}
	public void startHeartbeat() {
		new HeartbeatThread().start();
	}

	/**
	 * 负责注册的线程
	 * @author Azarias_Jx
	 *
	 */
	class RegisterThread extends Thread {
		
		CountDownLatch latch;
		
		public RegisterThread() {
		}
		
		@Override
		public void run() {
			try {
				// 发送rpc接口调用请求到NameNode去进行注册
				System.out.println("发送RPC请求到NameNode进行注册.......");  
				
				// 在这里进行注册的时候会提供哪些信息过去呢？
				// 比如说当前这台机器的ip地址、hostname，这两个东西假设是写在配置文件里的
				// 我们写代码的时候，主要是在本地来运行和测试，有一些ip和hostname，就直接在代码里写死了
				// 大家后面自己可以留空做一些完善，你可以加一些配置文件读取的代码
				String ip = "127.0.0.1";
				String hostname = "dfs-data-01";
				// 通过RPC接口发送到NameNode他的注册接口上去

				RegisterRequest registerRequest =
						RegisterRequest.newBuilder()
						.setIp(ip)
						.setHostname(hostname)
						.build();

				 RegisterResponse response = namenode_rpc.register(registerRequest);

				System.out.println("接收到NameNode返回的注册响应：" + response.getStatus());
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
	}
	/**
	 * 负责心跳的线程
	 */
	class HeartbeatThread extends Thread{
		@Override
		public void run() {
			while (true){
				String ip = "127.0.0.1";
				String hostname = "dfs-data-01";
				// 通过RPC接口发送到NameNode他的注册接口上去

				HeartbeatRequest heartbeatRequest =
						HeartbeatRequest.newBuilder()
								.setIp(ip)
								.setHostname(hostname)
								.build();

				HeartbeatResponse response = namenode_rpc.heartbeat(heartbeatRequest);
				try {
					Thread.sleep(1000*60);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				System.out.println("接受到namenode返回的心跳响应" + response.getStatus());
			}
		}
	}
	
}
